package com.giw.project.simple_graph_traversal;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.giw.project.exception_manager.UncheckedException;

public class TraversalReducer extends Reducer<Text, Text, Text, Text> {

	private final static String tmp_path = "tmp2.txt";
	private final static String tmp_path1 = "tmp.txt";
	private final static String tmp_pathS = "tmp_searchKind.txt";
	
	private final static Map<String, String> link2successors_director = new HashMap<String, String>();
	private final static Map<String, String> link2successors_genre =new HashMap<String, String> ();
	private final static Map<String, String> link2kind = new HashMap<String, String>();
	private final static Map<String, String> link2successors_actors = new HashMap<String, String>();
	private final static Map<String, Map<String,String>> kind2map = new HashMap<String, Map<String,String>>();

	protected void setup(Context context) {
		link2successors_director.put("title-<http://purl.org/dc/terms/title>",
				"<http://data.linkedmdb.org/resource" + "/movie/director>");
		link2successors_director.put("movie-<http://data.linkedmdb.org/resource"
				+ "/movie/director>",
				"<http://data.linkedmdb.org/resource/movie/director>");
		link2successors_director.put("director-<http://data.linkedmdb.org/resource"
				+ "/movie/director>", "<http://xmlns.com/foaf/0.1/page>");
		
		link2successors_genre.put("title-<http://purl.org/dc/terms/title>",
				"<http://data.linkedmdb.org/resource/movie/genre>");
		link2successors_genre.put("movie-<http://data.linkedmdb.org/resource"
				+ "/movie/genre>",
				"<http://data.linkedmdb.org/resource/movie/genre>");
		link2successors_genre.put("genre-<http://data.linkedmdb.org/resource"
				+ "/movie/genre>", "<http://xmlns.com/foaf/0.1/page>");
		
		link2successors_actors.put("title-<http://purl.org/dc/terms/title>",
				"<http://data.linkedmdb.org/resource/movie/actor>");
		link2successors_actors.put("movie-<http://data.linkedmdb.org/resource"
				+ "/movie/actor>",
				"<http://data.linkedmdb.org/resource/movie/actor>");
		link2successors_actors.put("actor-<http://data.linkedmdb.org/resource"
				+ "/movie/actor>", "<http://xmlns.com/foaf/0.1/page>");

		link2kind.put("<http://data.linkedmdb.org/resource/film/", "movie");
		link2kind.put("<http://data.linkedmdb.org/resource/director/",
				"director");
		link2kind.put("<http://data.linkedmdb.org/resource/film_genre/", "genre");
		link2kind.put("string", "title");
		link2kind.put("<http://data.linkedmdb.org/resource/actor/", "actor");
		
		kind2map.put("genre",link2successors_genre);
		kind2map.put("director",link2successors_director);
		kind2map.put("actor",link2successors_actors);
	}
	
	private Map<String,String> getSuccessorsMap() throws IOException{
		File f=new File(tmp_pathS);
		FileInputStream fis=new FileInputStream(f);
		InputStreamReader is=new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line=buffer.readLine();
		if(line!=null)
			return kind2map.get(line);
		else
			throw new UncheckedException("Error while trying to discover the kind of user search.\n");
		
	}

	protected void reduce(Text key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		
		Map<String, String> currentNodes2link = new HashMap<String, String>();
		File f = new File(tmp_path1);
		FileInputStream fis = new FileInputStream(f);
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line = buffer.readLine();
		Map<String,Integer> movie2rank = new HashMap<String,Integer>();
		while (line != null) {
			String currentNode=line.split("\t")[0];
			String currentLink=line.split("\t")[1];
			currentNodes2link.put(currentNode, currentLink);
			Integer count=movie2rank.get(currentNode);
			if(count==null){
				count=0;
			}
			count++;
			movie2rank.put(currentNode,count);
			line = buffer.readLine();
		}
		boolean endSearch = this.verifyEndSearch(currentNodes2link);
		buffer.close();
		is.close();
		fis.close();
		String node = key.toString();
		String link = currentNodes2link.get(node);
		if (link == null)
			link = currentNodes2link.get(node.substring(1, node.length() - 1));
		for (Text neighbour : values) {
			String[] neighbours = neighbour.toString().split("},");
			for (String neigh : neighbours) {
				if (!(neigh.length() < link.length() + 1)) {
					String current_link = neigh.substring(1, link.length() + 1);
					if (current_link.equals(link)) {
						try {
							String new_node = neigh.split(",")[1];
							if (new_node.charAt(new_node.length() - 1) == '.') {
								new_node = new_node.substring(0, new_node
										.length() - 2);
							}
							if (!endSearch) {
								String kind = this.getNodeKind(node);
								String a = kind + "-" + link;
								scriviSuFile(new_node, this.getSuccessorsMap().get(a));
							} else {
								context.write(new Text(node+"\t"+movie2rank.get(node)),
										new Text(new_node));
								File fS=new File(tmp_pathS);
								fS.delete();
							}
						} catch (Exception e) {
							throw new UncheckedException(
									"Error while trying to write the temporary file!\n"
											+ "Message:" + e.getMessage());
						}
					}
				}
			}
		}
	}

	private void scriviSuFile(String node, String link) throws Exception {
		File f = new File(tmp_path);
		FileOutputStream fos = null;
		if (f.exists()) {
			fos = new FileOutputStream(f,true);
		} else
			fos = new FileOutputStream(f);
		PrintStream output = new PrintStream(fos);
		output.println(node + "\t" + link);
		output.close();
		fos.close();

	}

	private boolean verifyEndSearch(Map<String, String> nodes2link) {
		for (String node : nodes2link.keySet()) {
			if (node.length() < 41)
				return false;
			if (node.substring(0, 41).equals(
					"<http://data.linkedmdb.org/resource/film/")
					&& nodes2link.get(node).equals(
							"<http://xmlns.com/foaf/0.1/page>")) {
				return true;
			}
			return false;
		}
		return false;
	}

	private String getNodeKind(String uri) {
		if (!uri.contains("http")) {
			return link2kind.get("string");
		}
		String[] parts_uri = uri.split("/");
		int i;
		uri = "";
		for (i = 0; i < parts_uri.length - 1; i++) {
			uri += "/" + parts_uri[i];
		}
		return link2kind.get(uri.substring(1) + "/");

	}

}
